package cn.itcast.map;

import cn.itcast.bean.StockBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.RichMapFunction;

import java.sql.Timestamp;
import java.util.Date;

public class StockPutHdfsMap extends RichMapFunction<StockBean,String> {
    /**
     * 开发步骤:
     *  1.定义字符串字段分隔符
     *  2.日期转换和截取：date类型
     *  3.新建字符串缓存对象
     *  4.封装字符串数据
     */
    private String split = QuotConfig.config.getProperty("hdfs.seperator");

    @Override
    public String map(StockBean stockBean) throws Exception {
        //todo 1.定义字符串字段分隔符
        //todo 2.日期转换和截取：date类型
        String tradeDate = DateUtil.longTimeToString(stockBean.getEventTime(), Constant.format_yyyy_mm_dd);
        //todo 3.新建字符串缓存对象
        StringBuilder sb = new StringBuilder();
        //todo 4.封装字符串数据
        //Timestamp|date|secCode|secName|preClosePrice|openPirce|highPrice|
        //lowPrice|closePrice|tradeVol|tradeAmt|tradeVolDay|tradeAmtDay|source
        sb.append(new Timestamp(stockBean.getEventTime())).append(split)
                .append(tradeDate).append(split)
                .append(stockBean.getSecCode()).append(split)
                .append(stockBean.getSecName()).append(split)
                .append(stockBean.getPreClosePrice()).append(split)
                .append(stockBean.getOpenPrice()).append(split)
                .append(stockBean.getHighPrice()).append(split)
                .append(stockBean.getLowPrice()).append(split)
                .append(stockBean.getClosePrice()).append(split)
                .append(stockBean.getTradeVol()).append(split)
                .append(stockBean.getTradeAmt()).append(split)
                .append(stockBean.getTradeVolDay()).append(split)
                .append(stockBean.getTradeAmtDay()).append(split)
                .append(stockBean.getSource());
        return sb.toString();
    }
}
